package com.settlement.system.service.csv;

import com.settlement.system.csv.CsvDatas;
import com.settlement.system.csv.CsvWrite;
import com.settlement.system.common.exception.ServiceException;
import com.settlement.system.service.IBaseService;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author author
 */
@Slf4j
public class BaseServiceCsvDatas implements CsvDatas {

    private static ExecutorService executorService = Executors.newFixedThreadPool(8);

    protected IBaseService baseService;
    protected Map<String, Object> params;
    private AtomicInteger uncompleteTaskNum;
    private BlockingQueue<Object> queue = new LinkedBlockingQueue<>(100000);
    /**
     * 是否发生异常
     */
    private volatile Boolean isException = false;

    public BaseServiceCsvDatas(IBaseService baseService, Map<String, Object> params) {
        this.baseService = baseService;
        this.params = params;
    }

    @Override
    public void pushDatasToCsvWrite(CsvWrite csvWrite) throws IOException {
        List<Map<String, Object>> listParams = getListParams(params);
        uncompleteTaskNum = new AtomicInteger(listParams.size());

        for (Map<String, Object> objectMap : listParams) {
            String ifMapData = (String)params.get("ifMapData");
            if("true".equalsIgnoreCase(ifMapData)){
                executorService.execute(new PushMapDataToQueueTask(objectMap));
            }else{
                executorService.execute(new PushDataToQueueTask(objectMap));
            }

        }
        pollQueueDataToCsvWrite(csvWrite);

        if (isException) {
            throw new IOException("写入数据异常");
        }
    }

    /**
     * 搜索条件的拆分,子类可以按照自己规则来拆分
     *
     * @return
     */
    protected List<Map<String, Object>> getListParams(Map<String, Object> params) {
        return Arrays.asList(params);
    }

    /**
     * 推送数据到阻塞队列中
     */
    private class PushMapDataToQueueTask implements Runnable {

        Map<String, Object> taskParams;

        public PushMapDataToQueueTask(Map<String, Object> params) {
            this.taskParams = params;
        }

        @Override
        public void run() {
            try {

                if (isException) {
                    return;
                }

                baseService.getMapDataByHandler(taskParams, resultContext -> {

                    if (isException) {
                        resultContext.stop();
                    }

                    try {
                        Map<String, Object> resultObject = resultContext.getResultObject();
                        if(resultObject==null){
                            resultObject = new HashMap<String,Object>();
                        }
                        queue.put(beforeWrite(resultObject));
                    } catch (Exception e) {
                        isException = true;
                        log.error("获取数据发生中断", e);
                    }
                });
            } catch (ServiceException e) {
                log.error("获取数据发生异常", e);
                isException = true;
            } finally {
                uncompleteTaskNum.decrementAndGet();
            }

        }
    }

    /**
     * 推送数据到阻塞队列中
     */
    private class PushDataToQueueTask implements Runnable {

        Map<String, Object> taskParams;

        public PushDataToQueueTask(Map<String, Object> params) {
            this.taskParams = params;
        }

        @Override
        public void run() {
            try {

                if (isException) {
                    return;
                }

                baseService.getDataByHandler(taskParams, resultContext -> {

                    if (isException) {
                        resultContext.stop();
                    }

                    try {
                        queue.put(beforeWrite(resultContext.getResultObject()));
                    } catch (InterruptedException e) {
                        isException = true;
                        log.error("获取数据发生中断", e);
                    }
                });
            } catch (ServiceException e) {
                log.error("获取数据发生异常", e);
                isException = true;
            } finally {
                uncompleteTaskNum.decrementAndGet();
            }

        }
    }

    private void pollQueueDataToCsvWrite(CsvWrite csvWrite) {

        while (true) {
            boolean flag = (queue.size() == 0 && uncompleteTaskNum.get() == 0) || isException;
            if (flag) {
                break;
            }

            try {
                Object data = queue.poll(100, TimeUnit.MILLISECONDS);
                if (data != null) {
                    csvWrite.write(data);
                }
            } catch (Exception ex) {
                ex.printStackTrace();
                isException = true;
            }
        }
    }

    /**
     * 写入文件之前执行
     *
     * @param data
     * @return
     */
    public Object beforeWrite(Object data) {
        return data;
    }
}
